package org.budo.support.service.batch.buffered;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.budo.support.spring.aop.util.AopUtil;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * 干啥用: 异步缓冲批量写数据, 适用于大量非全实时写入日志等数据
 * 如何使用 继承我; 调用 addToBuffer 加入缓冲;实现 flashBuffer 方法 批量写入
 * @author lmw
 */
@Slf4j
@Getter
@Setter
public abstract class AbstractBufferedBatchService<T> {
    /**
     * 用于配置executor的属性
     */
    private ExecutorService executorService;

    private Integer bufferSize = 100;

    private Integer executorThreads = 2;

    /**
     * 最大缓冲时间 一秒
     */
    private long bufferPeriod = 1 * 1000;

    /**
     * 如果未配置就生产默认的 并用这个属性缓存之
     */
    private ExecutorService _executorService;

    /**
     * 缓冲区
     */
    private List<T> _buffer = new ArrayList<T>();

    /**
     * 最后一次缓冲刷新时间
     */
    private long _lastFlashAt;

    /**
     * 刷新缓冲区 将数据批量写入
     */
    public abstract void flashBuffer(List<T> bufferCopy);

    public void addToBuffer(List<T> list) {
        if (null == list || list.isEmpty()) {
            log.error("#51 addToBuffer, this=" + this + ", list=" + list);
            return;
        }

        for (T each : list) {
            this.addToBuffer(each);
        }
    }

    public void addToBuffer(T entity) {
        if (null == entity) {
            log.error("#62 addToBuffer, this=" + this + ", entity=" + entity);
            return;
        }

        try {
            this._buffer.add(entity); // 加入缓冲

            // 缓冲区满了就刷新
            if (this._buffer.size() >= this.getBufferSize()) { // 写入数据库
                this.submitBuffer();
                return;
            }

            // 时间到了也刷新
            if ((System.currentTimeMillis() - this._lastFlashAt) > this.getBufferPeriod()) {
                this.submitBuffer();
            }
        } catch (Throwable e) {
            log.error("#35 this=" + this + ", addToBuffer error, " + e, e);
        }
    }

    private void submitBuffer() {
        final List<T> bufferCopy = new ArrayList<T>(this._buffer);
        this._buffer.clear();

        if (null == bufferCopy || bufferCopy.isEmpty()) {
            log.error("#72 submitBuffer, this=" + this + ", bufferCopy=" + bufferCopy);
            return;
        }

        this.executor().submit(new Runnable() {
            public void run() {
                try {
                    AbstractBufferedBatchService.this.flashBuffer(bufferCopy);
                } catch (Throwable e) { // 捕获并日志异步方法里的异常
                    log.error("#48 submitBuffer, this=" + this + ", _flashBuffer error, e=" + e, e);
                    throw e instanceof RuntimeException ? (RuntimeException) e : new RuntimeException(e);
                }
            }
        });

        this._lastFlashAt = System.currentTimeMillis();
    }

    public ExecutorService executor() {
        if (null != this.getExecutorService()) {
            return this.getExecutorService();
        }

        if (null != this.getExecutorThreads()) {
            if (null == this._executorService) {
                this._executorService = Executors.newFixedThreadPool(this.getExecutorThreads());
            }

            return this._executorService;
        }

        throw new RuntimeException("#68 executorService or executorThreads needs to be configured");
    }

    /**
     * 销毁前执行一遍
     */
    public void preDestroy() {
        log.warn("#116 preDestroy, this=" + this);

        AbstractBufferedBatchService<T> _this = AopUtil.proxy(this);
        _this.flashBuffer(this._buffer);

        this._buffer.clear();
    }
}
